/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.jms;

import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Vector;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TemporaryTopic;
import javax.jms.Topic;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;

import cgl.narada.util.UUIDRetriever;

/**
 * This class implements the TopicSession interface outlined in the JMS
 * specification.
 * 
 * @author Shrideep Pallickara $Revision$ $Date$
 */

public class JmsTopicSession extends JmsSession implements TopicSession,
    JmsDebugFlags {

    /** Indicates wether the session is transacted or not */
    private boolean transacted;

    /** The acknowledge mode */
    private int acknowledgeMode;

    /** The JMS TopicConnection that created this TopicSession */
    private JmsTopicConnection jmsTopicConnection;

    /** The list of managed publishers & subscribers registered to this session */
    private Vector managedPublishers, managedSubscribers;

    /** The list of durable subscribers */
    private Hashtable durableSubscribers;
    
    /** The module name for debugging purposes */
    private String moduleName = "JmsTopicSession: ";

    public JmsTopicSession(JmsTopicConnection jmsTopicConnection,
        boolean transact, int ackMode) {
        this.jmsTopicConnection = jmsTopicConnection;
        transacted = transact;
        acknowledgeMode = ackMode;

        managedPublishers = new Vector();
        managedSubscribers = new Vector();

        durableSubscribers = new Hashtable();
        if (JmsDebugFlags.JmsTopicSession_Debug)
            System.out.println("JMS TopicSession Initialized ...");
    }

    /**
     * Creates a durable subscriber to the specified topic.
     * 
     * @param topic
     *            the non-temporary Topic to subscribe to
     * @param name
     *            the name used to identify this subscription
     */
    public TopicSubscriber createDurableSubscriber(Topic topic, String name)
        throws JMSException {

        return createTopicSubscriber(topic, name, null/* no selector */, true/*
                                                                             * accept
                                                                             * local
                                                                             */);
    }

    public TopicSubscriber createDurableSubscriber(Topic topic, String name,
                                                   String messageSelector,
                                                   boolean noLocal)
        throws JMSException {

        return createTopicSubscriber(topic, name, messageSelector, noLocal);
    }

    public TopicPublisher createPublisher(Topic topic) throws JMSException {
        if (topic == null)
            throw new JMSException("The specified topic is Null");

        /** Proceed to instantiate the topicPublisher */
        JmsTopicPublisher topicPublisher = new JmsTopicPublisher(this, topic);
        managedPublishers.addElement(topicPublisher);

        /** ************ Anabas related start *********** */
        /*
         * createTopicSubscriber(topic,null,null,true); if
         * (JmsTopicSession_Debug) { System.out.println("\n" + "Created a
         * SUBSCRIBER TO the topic " + topic + " that the PUBLISHER publishes to " +
         * "\n"); }
         */
        /** ************ Anabas related end ************* */

        return topicPublisher;
    }

    public TopicSubscriber createSubscriber(Topic topic) throws JMSException {
        return createTopicSubscriber(topic, null /* not durable */,
                                     null /* no selector */, true /* accept local */);
    }

    public TopicSubscriber createSubscriber(Topic topic,
                                            String messageSelector,
                                            boolean noLocal)
        throws JMSException {
        // return createTopicSubscriber(topic, null /*not durable*/,
        // null /*no selector*/,true /*accept local */);
        return createTopicSubscriber(topic, null /* not durable */,
                                     messageSelector /* no selector */, true /*
                                                                             * accept
                                                                             * local
                                                                             */);
    }

    public Topic createTopic(String topicName) throws JMSException {
        return new JmsTopic(topicName);
    }

    public TemporaryTopic createTemporaryTopic() throws JMSException {
        // System.out.println("Requesting TEMPORARY TOPIC");
        UUIDRetriever retriever = UUIDRetriever.getInstance();
        String topicString = retriever.getRandomBasedUUIDAsString();
        return new JmsTopic(topicString);
        // return new JmsTemporaryTopic(this);
    }

    private synchronized TopicSubscriber createTopicSubscriber(
                                                               Topic topic,
                                                               String name,
                                                               String messageSelector,
                                                               boolean noLocal)
        throws JMSException {

        if (topic == null)
            throw new JMSException("The specified topic is null");

        if (name != null) {
            /** This is a durable subscription */
            if (durableSubscribers.containsKey(name))
                throw new JMSException("A durable subscription" + name
                    + " already exists");
        }

        /** Proceed to instantiate the TopicSubscriber */
        JmsTopicSubscriber topicSubscriber = new JmsTopicSubscriber(
                                                                    this,
                                                                    topic,
                                                                    name,
                                                                    messageSelector,
                                                                    noLocal);

        /**
         * Add this subscriber to the list of subscribers managed by the session
         */
        if (name == null)
            managedSubscribers.addElement(topicSubscriber);
        else
            durableSubscribers.put(name, topicSubscriber);

        return topicSubscriber;
    }

    /**
     * Unsubscribes a durable subscription that has been created by a client. It
     * is erroneous for a client to delete a durable subscription while there is
     * an active TopicSubscriber for the subscription, or while a consumed
     * message is part of a pending transaction or has not been acknowledged in
     * the session
     */
    public void unsubscribe(String name) throws JMSException {
        JmsTopicSubscriber topicSubscriber = null;
        if (durableSubscribers.containsKey(name)) {
            topicSubscriber = (JmsTopicSubscriber)durableSubscribers.remove(name);
        }/* end-if */

        /** Proceed to eliminate the durable subscription */
        /** JMS-NI */

        if (topicSubscriber == null) {
            /** Try to remove a regular subscriber: based on the topic, instead
             * of the name of the subscription as in the durable case */
            topicSubscriber = removeSubscriber(name);
        }

        
        if (topicSubscriber == null) {
            throw new JMSException("The subscription (" + name
                                   + ") does not exist");
        }
        
        topicSubscriber.unSubscribe();
    }

    
    private JmsTopicSubscriber removeSubscriber(String subscription) 
         throws JMSException {
        for (int i = 0; i < managedSubscribers.size(); i++) {
            JmsTopicSubscriber jmsTSub = (JmsTopicSubscriber) managedSubscribers
                .elementAt(i);
            if (jmsTSub.getTopic().getTopicName().equals(subscription)) {
                managedSubscribers.removeElementAt(i);
                return jmsTSub;
            }
        }/* end-for */
        
        return null;
    }
    
    
    public JmsTopicConnection getJmsTopicConnection() {
        return jmsTopicConnection;
    }

    public void routeEventToAppropriateSubscribers(JmsEvent jmsEvent) {
        for (int i = 0; i < managedSubscribers.size(); i++) {
            JmsTopicSubscriber jmsTSub = (JmsTopicSubscriber) managedSubscribers
                .elementAt(i);
            checkAndRouteToSubscriber(jmsEvent, jmsTSub);
        }/* end-for */

        /** Check durable subscribers now */
        Enumeration keys = durableSubscribers.keys();
        while (keys.hasMoreElements()) {
            JmsTopicSubscriber jmsTSub = (JmsTopicSubscriber) durableSubscribers
                .get(keys.nextElement());
            checkAndRouteToSubscriber(jmsEvent, jmsTSub);
        }
    }

    private synchronized void checkAndRouteToSubscriber(
                                                        JmsEvent jmsEvent,
                                                        JmsTopicSubscriber jmsTSub) {
        if (jmsEvent == null) {
            System.out.println(moduleName + "The JMSEvent is NULL");
            return;
        }

        if (jmsEvent.getMessage() == null) {
            System.out.println(moduleName + "ENcapsulated message is NULL");
            return;
        }
        if (jmsTSub == null) {
            System.out.println(moduleName
                + "JmsTSub ... topic subscriber is null");
            return;
        }

        try {
            // String topicInEvent = jmsEvent.getTopic().getTopicName();
            JmsTopic eventTopic = (new JmsTopic(jmsEvent.getTopic()));
            JmsTopic tSubTopic = ((JmsTopic) jmsTSub.getTopic());
            if (eventTopic == null || tSubTopic == null) {
                System.out.println("\n \n" + moduleName
                    + "Unmarshalled topic/tSubTopic NULL");
                return;
            }
            String topicInEvent = eventTopic.getTopicName().trim();
            String selector = jmsTSub.getTopic().getTopicName().trim();

            if (selector == null || topicInEvent == null) {
                System.out.println(moduleName + "Topic string is null");
                return;
            }

            /** This just indicates thet the Topic has been matched so far */
            boolean topicMatched = false;
            if (topicInEvent.equals(selector)) {
                topicMatched = true;
            }

            /*
             * boolean topicMatched =
             * JmsReceivedEventDistributor.matchesSpecifiedSelector(topicInEvent,
             * selector);
             */

            /**
             * Once this topic is matched, we need to confirm that this
             * subscriber should indeed receive the event
             */
            boolean routeToSub = false; /*
                                         * Indicates whether event should go to
                                         * the subscriber
                                         */
            if (topicMatched) {
                JmsSelector _jmsSelector = jmsTSub.getJmsSelector();
                if (_jmsSelector == null) {
                    routeToSub = true;
                    /** Since no message selector was specified */
                } else {
                    routeToSub = _jmsSelector.eventMatches(jmsEvent
                        .getMessage());
                } /* end if - for selector */
            }/* end if (topicMatch) */

            if (routeToSub) {
                if (JmsTopicSession_Debug) {
                    System.out
                        .println(">>>> Found valid subscriber to route event "
                            + "to <<<<");
                }/* end-debug printing */
                if (jmsEvent == null) {
                    System.out.println(moduleName + "JmsEvent == null");
                    return;
                }

                Message jmsMsg = jmsEvent.getMessage();
                if (jmsMsg == null) {
                    System.out.println(moduleName + " JmsMessage is null");
                    return;
                }
                if (jmsTSub == null) {
                    System.out.println(moduleName + " TSub is null");
                    return;
                }
                MessageListener listener = jmsTSub.getMessageListener();
                if (listener == null) {
                    System.out
                        .println(moduleName + " Message listener is null");
                    return;
                }

                if (jmsMsg.getJMSType().equals("TextMessage")) {
                    // System.out.println(moduleName + jmsMsg);
                }

                /** Added to deal with Pierce Group issue */
                if (jmsMsg.getJMSType().equals("BytesMessage")) {
                    JmsBytesMessage bytesMsg = (JmsBytesMessage) jmsMsg;
                    bytesMsg.reset();
                }
                /** Pierce group issue end */

                if (jmsMsg.getJMSType().equals("StreamMessage")) {
                    JmsStreamMessage streamMsg = (JmsStreamMessage) jmsMsg;
                    streamMsg.reset();
                }

                // jmsTSub.getMessageListener().onMessage(jmsEvent.getMessage());
                listener.onMessage(jmsMsg);

            }

        } catch (JMSException jmsEx) {
            System.out.println("JmsTopicSession:routeEventToSubs:: " + jmsEx);
        }/* end try-catch block */
    }

}
